package com.ruoyi.mqtt.service;

import com.alibaba.fastjson2.JSON;
import com.ruoyi.mqtt.pojo.Mqtt;
import com.ruoyi.mqtt.pojo.MqttLogs;
import org.apache.tomcat.util.threads.ThreadPoolExecutor;
import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttCallback;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import javax.websocket.*;
import javax.websocket.server.ServerEndpoint;
import java.io.IOException;
import java.time.LocalDateTime;
import java.util.Set;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.CopyOnWriteArraySet;
import java.util.concurrent.TimeUnit;


@Component
public class MqttMessageCallback1 implements MqttCallback {


    @Resource
    private RedisTemplate redisTemplate;

    @Resource
    private MqttLogsService mqttService;

    @Resource
    private HBaseService hBaseService;




    /**
     *  PostConstruct注解的方法。在Spring框架中使用该注解标记的方法会在依赖注入完成后自动执行，通常用于初始化工作。
     * @author songshijia
     * @date 2023/11/20
     */
    private ThreadPoolExecutor threadPoolExecutor = null;
    @PostConstruct
    public void setMqttMessageCallback1(){
        threadPoolExecutor = new ThreadPoolExecutor(
                5,
                10,
                30, TimeUnit.SECONDS,
                new ArrayBlockingQueue<>(15),
                new ThreadPoolExecutor.AbortPolicy()
        );
    }

    /**
     *
     * MQTT 断开连接会执行此方法
     */
    @Override
    public void connectionLost(Throwable throwable) {
        System.out.println("断开了MQTT连接 ：" + throwable.getMessage());
        System.out.printf(throwable.getMessage(), throwable);
    }

    /**
     * publish发布成功后会执行到这里
     */
    @Override
    public void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) {
        System.out.printf("发布消息成功\n");
    }


    /**
     * subscribe订阅后得到的消息会执行到这里
     */
    @Override
    public void messageArrived(String topic, MqttMessage message) throws Exception {
        //  TODO    此处可以将订阅得到的消息进行业务处理、数据存储
        System.out.println("开始"+ LocalDateTime.now());
        try {
            threadPoolExecutor.submit(()->{

                String content = new String(message.getPayload());
                System.out.println("收到来自 " + topic + " 的消息："+ content);


                //将json转为对象
                MqttLogs logs = JSON.parseObject(content, MqttLogs.class);
                Mqtt mqtt = JSON.parseObject(content, Mqtt.class);
                hBaseService.save(logs);



                MqttLogs mqttLogs = new MqttLogs();
                mqttLogs.setVin(mqtt.getVin());
                mqttLogs.setTopic(mqtt.getTopic());
                mqttLogs.setTemperature(mqtt.getTemperature());
                mqttLogs.setHumidity(mqtt.getHumidity());
                mqttLogs.setMaxTemperature(mqtt.getMaxTemperature());
                mqttLogs.setMinTemperature(mqtt.getMinTemperature());
                mqttLogs.setMaxHumidity(mqtt.getMaxHumidity());
                mqttLogs.setMinHumidity(mqtt.getMinHumidity());
                mqttLogs.setTime(LocalDateTime.now());


                //T001  H001  TH001
                Boolean T = false,H = false;
                if (mqttLogs.getTemperature() > mqttLogs.getMaxTemperature() ||
                    mqttLogs.getTemperature() < mqttLogs.getMinTemperature()){
                    System.err.println("温度异常");
                    mqttLogs.setAbnormal("T");
                    T = true;
                }
                if (mqttLogs.getHumidity() > mqttLogs.getMaxHumidity() ||
                    mqttLogs.getHumidity() < mqttLogs.getMinHumidity()){
                    System.err.println("湿度异常");
                    mqttLogs.setAbnormal("H");
                    H = true;
                }

                if (T && H){
                    mqttLogs.setAbnormal("TH");
                }

                if (T || H){
                    //加入Redis缓存
                    MqttLogs redisMqtt = (MqttLogs) this.redisTemplate.opsForValue().get("REDISMQTT" + mqttLogs.getVin());
                    if (redisMqtt!=null){
                        if ( mqttLogs.getTemperature() <= redisMqtt.getTemperature()+0.1   &&
                                mqttLogs.getTemperature() >= redisMqtt.getTemperature()-0.1  &&
                                mqttLogs.getHumidity() <= redisMqtt.getHumidity()+0.1   &&
                                mqttLogs.getHumidity() >= redisMqtt.getHumidity()-0.1)
                        {
                            System.out.println("温度湿度一至");
                        }else {
                            this.redisTemplate.opsForValue().set("REDISMQTT" + mqttLogs.getVin(),mqttLogs,5,TimeUnit.MINUTES);
                            System.out.println("数据不一致,重新加入数据库");
                            this.mqttService.insertTbMqttLogs(mqttLogs);
                        }

                    }else{
                        this.redisTemplate.opsForValue().set("REDISMQTT" + mqttLogs.getVin(),mqttLogs,5,TimeUnit.MINUTES);
                        System.out.println("加入数据库");
                        this.mqttService.insertTbMqttLogs(mqttLogs);
                    }

                }
                    System.out.println("数据处理完毕:"+LocalDateTime.now()+"Thread Name:"+Thread.currentThread().getName());

            });
        } catch (Exception e) {
            e.printStackTrace();
        }
        System.out.println("结束"+LocalDateTime.now());
    }



}
